#include "StdAfx.h"
#include "JsonStreamServer.h"
#include "json.h"
#include "MemoryManager.h"
#include "Types.h"
#include "Element.h"
#include "TimestampGenerator.h"
#include "DocumentIdentifierGenerator.h"
#include "QueueEntity.h"
#include "RelationSynopsis.h"
#include "LineageSynopsis.h"
#include "JsonStreamServer.h"
#include "JsonSchema.h"
#include "SchemaInterpreter.h"
#include "Operator.h"
#include "RowWindowOperator.h"
#include "JoinOperator.h"
#include "SelectionOperator.h"
#include "ProjectionOperator.h"
#include "RootOperator.h"
#include "RstreamOperator.h"
#include "IstreamOperator.h"
#include "DstreamOperator.h"
#include "LeafOperator.h"
#include "Scheduler.h"
#include "IOManager.h"
#include "SocketStreamInput.h"
#include "IStreamInput.h"
#include "RandomGeneratedStreamInput.h"
#include "ShowResultStreamOutput.h"
#include "IStreamOutput.h"
#include "QueryManager.h"
#include "QueryIntermediateRepresentation.h"
#include "JsonSchema.h"
#include "PlanManager.h"
#include "SchemaManager.h"
#include "CommandManager.h"
#include "ConfigureManager.h"
#include "WrapperManager.h"
#include <windows.h>
#include <boost/thread/condition.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>
JsonStreamServer* JsonStreamServer::jsonStreamServer = NULL;
JsonStreamServer::JsonStreamServer(void)
{
	initial(); //initial the server 
}


JsonStreamServer::~JsonStreamServer(void)
{

}
JsonStreamServer* JsonStreamServer:: getInstance(void)
{
	if(jsonStreamServer == NULL)
	{
		jsonStreamServer = new JsonStreamServer();
	}
	return jsonStreamServer;
}
/* initial each module manager in the system*/
void JsonStreamServer::initial()
{

	isDataComing = false;

	std::cout<<"initial"<<std::endl;
	ConfigureManager::getInstance(); // read configuration file, this manager should be initialed first
	std::cout<<"configuration initialed"<<std::endl;
	IOManager::getInstance();        // deal with IO socket 
	std::cout<<"IOManager initialed"<<std::endl;
	MemoryManager::getInstance();    // allocate and manage memory for queues and synopsis 
	std::cout<<"MemoryManager initialed"<<std::endl;
	PlanManager::getInstance();      // generate query plan and instantiate operators by query intermediate representation
	std::cout<<"PlanManager initialed"<<std::endl;
	QueryManager::getInstance();     // manage the input and output of the query registered
	std::cout<<"QueryManager initialed"<<std::endl;
	SchemaManager::getInstance();    // register schema information
	std::cout<<"SchemaManager initialed"<<std::endl;
	CommandManager::getInstance();   // system listens to a port waiting for command 
	std::cout<<"CommandManager initialed"<<std::endl;
	WrapperManager::getInstance();   // read the wrapper specification in the wrapper folder, and register each wrapper 
	std::cout<<"WrapperManager initialed"<<std::endl;

	
}


/* server execution, each time get one operator and the execution times from the scheduler */
void JsonStreamServer::execute()
{
	LARGE_INTEGER  large_interger;  
	double dff;  
	long long  start = -1, end = -1;  
	QueryPerformanceFrequency(&large_interger);  
	dff = large_interger.QuadPart;  

	std::cout<<"begin executing"<<std::endl;
	//for(int i =0;i<5000000;i++)
	int i = 0;
	IOManager::getInstance()->execute();
	while(1)
	{
		//boost::mutex::scoped_lock lk(monitorMutex);
		//while (isDataComing == false)
		//	isDataComingCondition.wait(lk);
		//isDataComing = true;

		 
		
		boost::shared_ptr<Operator> op;
		int executionTimes;
		if(!Scheduler::getInstance()->isEmpty()) // check if there is an operator waiting to execute
		{ //yes
			if(start<0)
			{
				QueryPerformanceCounter(&large_interger);  
				start = large_interger.QuadPart;  
				std::cout<<"receiving data..."<<std::endl;
			}

			Scheduler::getInstance()->getNextOperatorToExecute(op,executionTimes);
			op->execution(executionTimes); //execute the operator

		}
		else 
		{//release CPU resource
			i--;
			//Sleep(5000);
		}
	}
	std::cout<<"finished executing"<<std::endl;
	QueryPerformanceCounter(&large_interger);  
	end = large_interger.QuadPart;  
	double result=(end - start)  / dff;
	int throughout = RandomGeneratedStreamInput::throughout;
	int total_dequeue_number = QueueEntity::total_dequeue_number;
	std::cout<<"total second :  "<<result<<std::endl;
	std::cout<<"throughtout tuple /per second: "<<throughout/result<<std::endl;
	std::cout<<"dequeue     tuple /per second: "<<total_dequeue_number/result<<std::endl;

}

void JsonStreamServer::registerQuery(boost::shared_ptr<QueryIntermediateRepresentation>intermediateQuery,boost::shared_ptr<IStreamOutput> streamOutput)
{
	QueryManager::getInstance()->addStreamOutput(streamOutput); //register query
	PlanManager::getInstance()->generateQueryPlan(intermediateQuery); // generate plan
	Scheduler::getInstance()->setOperatorList(PlanManager::getInstance()->getOperatorList());// inform scheduler
}
void JsonStreamServer::registerStreamInput(boost::shared_ptr<IStreamInput> streamInput)
{
	QueryManager::getInstance()->addStreamInput(streamInput); //register query
}